package com.yangye.niodemo.multiplexing.threadgroup;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yangye
 * @date 2020/12/24
 */
public class SelectorThreadGroup {

	public SelectorThread[] sts;

	public SelectorThreadGroup worker;

	public AtomicInteger xid = new AtomicInteger();

	public void setWorker(SelectorThreadGroup stg) {
		this.worker = stg;
	}

	public SelectorThreadGroup(int threadNum) {
		sts = new SelectorThread[threadNum];
		for (int i = 0; i < threadNum; i++) {
			sts[i] = new SelectorThread(this);
			new Thread(sts[i]).start();
		}
	}

	public void bind(int port) {
		try {
			ServerSocketChannel server = ServerSocketChannel.open();
			server.configureBlocking(false);
			server.bind(new InetSocketAddress(port));
			// 注册到selector上
			registerBoss(server);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void registerBoss(ServerSocketChannel serverSocketChannel) {
		SelectorThread st = this.sts[0];
		st.queue.add(serverSocketChannel);
		st.selector.wakeup();
	}

	public void registerWorker(SocketChannel socketChannel) {
		// 轮询选出下一个worker
		SelectorThread st = nextWorker();
		// 添加到队列
		st.queue.add(socketChannel);
		// 唤醒阻塞的多路复用器
		st.selector.wakeup();
	}

	public SelectorThread nextWorker() {
		int index = xid.get() % worker.sts.length;
		xid.getAndIncrement();
		return worker.sts[index];
	}
}
